package graph

import Configer.Configer
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import portrait._
import utils.GetTagsUserid

import scala.collection.mutable.ListBuffer

//图计算用户画像
object GraphTags {
  def main(args: Array[String]): Unit = {
    //saprkContext
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(s"${this.getClass.getName}")
      .set("spark.serializer",Configer.serializer)
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\parquet")
    val appLines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\app_dict.txt")
    val appMap = appLines.map(_.split("\t",-1)).filter(_.length>=5).map(arr=>(arr(4),arr(1))).collectAsMap()
    val stopLines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\stopwords.txt")
    val stopMap = stopLines.map((_,null)).collectAsMap()

    //广播变量
    val appBro = sc.broadcast(appMap)
    val stopBro = sc.broadcast(stopMap)
    //处理数据
    //点集合
    val pointRDD: RDD[(Long, List[(String, Int)])] = dataFrame.filter(GetTagsUserid.abc)
      .flatMap(row => {
        //
        val adList = adTags.makeTags(row)
        //
        val appList = appTags.makeTags(row, appBro)
        //
        val busList = BusinessTags.makeTags(row)
        //
        val deList = DeviceTags.makeTags(row)
        //
        val keyList = KeyTags.makeTags(row, stopBro)
        //
        val userAllId: ListBuffer[String] = GetTagsUserid.getUserid(row)
        val idList: ListBuffer[(String, Int)] = userAllId.map((_, 0))
        val allList = adList ++ appList ++ busList ++ deList ++ keyList ++ idList
        userAllId.map(id => {
          if (id.equals(userAllId.head)) {
            (id.hashCode().toLong, allList)
          } else {
            (id.hashCode.toLong, Nil)
          }
        })
      })
    //边集合
    val edgeRDD: RDD[Edge[Int]] = dataFrame.filter(GetTagsUserid.abc)
      .flatMap(row => {
        val userAllId: ListBuffer[String] = GetTagsUserid.getUserid(row)
        userAllId.map(id => Edge(userAllId.head.hashCode.toLong, id.hashCode.toLong, 0))
      })
    edgeRDD
    //图计算实例
    val graph = Graph(pointRDD,edgeRDD)
    val ver = graph.connectedComponents().vertices
    ver.join(pointRDD).map{
      case (id,(mid,list)) => (mid,list)
    }.reduceByKey{
      (list1,list2) => (list1++list2).groupBy(_._1)//.mapValues(tp=>tp.length).toList
          //.mapValues(tp=>tp.map(_._2).sum).toList
          .mapValues(tp=>tp.foldLeft(0)(_+_._2)).toList
    }.coalesce(1).saveAsTextFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\tags")

    //释放资源
    sc.stop()
  }
}
